package com.cn.common.consumer;

import com.cn.common.ConsumeQueueData;
import com.cn.common.config.ConfigManager;
import com.cn.util.GsonUtil;
import org.apache.commons.lang3.StringUtils;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class ConsumerOffsetManager extends ConfigManager {

    /**
     * 文件名
     */
    private String path;
    private ConcurrentMap<String, ConcurrentMap<Integer, Long>> offsetTables = new ConcurrentHashMap<>();

    public ConsumerOffsetManager(String path) {
        this.path = path;
        load();
    }

    //查询位移
    public ConcurrentMap<Integer, Long> readOffset(String topic) {
        return this.offsetTables.get(topic);
    }

    //查询位移
    public long readOffset(String topic, int queueid) {
        ConcurrentMap<Integer, Long> offsets = this.offsetTables.get(topic);
        if (offsets != null) {
            Long offset = offsets.get(queueid);
            if (offset != null) {
                return offset;
            }
        }
        return 0l;
    }

    public void updateOffset(ConsumeQueueData consumeQueueData) {
        ConcurrentMap<Integer, Long> offset = this.offsetTables.get(consumeQueueData.getTopic());
        if (null == offset) {
            offset = new ConcurrentHashMap<>();
            this.offsetTables.put(consumeQueueData.getTopic(), offset);
        }
        if (offset != null) {
            offset.put((int) consumeQueueData.getQueueId(), consumeQueueData.getOffset());
        }
    }

    //持久化到硬盘
    public void persit() {
        ConsumerOffset consumerOffset = new ConsumerOffset();
        consumerOffset.getTables().add(this.offsetTables);
        this.string2File(this.path, GsonUtil.converObj2Json(consumerOffset));
    }

    public void load() {
        String result = this.readString(this.path);
        if (StringUtils.isEmpty(result)) {
            return;
        }
        ConsumerOffset consumerOffset = new ConsumerOffset();
        consumerOffset = (ConsumerOffset) GsonUtil.convertJson2Obj(result, consumerOffset);
        if (consumerOffset != null && consumerOffset.getTables() != null) {
            for (ConcurrentMap<String, ConcurrentMap<Integer, Long>> map : consumerOffset.getTables()) {
                for (String key : map.keySet()) {
                    this.offsetTables.put(key, map.get(key));
                }
            }
        }
    }

    public static void main(String[] args) {
        String path = "C:\\F\\data3\\test\\demo.json";
        ConsumerOffsetManager test = new ConsumerOffsetManager(path);
        test.load();

//        ConcurrentMap<Integer, Long> son1 = new ConcurrentHashMap<>();
//        son1.put(1, 0l);
//        son1.put(2, 3l);
//
//        ConcurrentMap<Integer, Long> son2 = new ConcurrentHashMap<>();
//        son2.put(1, 0l);
//        son2.put(2, 8l);
//
//        test.offsetTables.put("top1", son1);
//        test.offsetTables.put("top2", son2);
//
//        test.persit();

    }

}
